package com.chenjj.bigdata.flink.watermark

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

/**
  *  根据时间周期水位线生成策略
  */
class PeriodicAssigner extends  AssignerWithPeriodicWatermarks[(String,Long,Int)]{
  val maxOutOfOrderness = 1000L//1秒时延设定，表示崽1秒以内的数据延时有效，超过1秒的数据呗认为迟到事件

  var currentMaxTimestamp:Long = _

  //获取当前事件时间
  override def extractTimestamp(element: (String, Long, Int), previousElementTimestamp: Long): Long = {
    //当前事件时间
    val currentTimestamp = element._2
    //对比当前事件时间和历史最大事件时间，将最新的时间赋值给currentMaxTimestamp
    currentMaxTimestamp = Math.max(currentTimestamp,currentMaxTimestamp)
    currentMaxTimestamp
  }

  override def getCurrentWatermark: Watermark = {
    new Watermark(currentMaxTimestamp-maxOutOfOrderness)
  }
}
